#ifndef __M_CHANNEL_H__
#define __M_CHANNEL_H__
#include "../mqcommon/mq_logger.hpp"
#include "../mqcommon/mq_helper.hpp"
#include "../mqcommon/mq_msg.pb.h"
#include "mq_consumer.hpp"
#include "mq_host.hpp"
#include <iostream>
#include "muduo/proto/codec.h"
#include "muduo/proto/dispatcher.h"
#include "muduo/net/TcpConnection.h"
#include "../mqcommon/mq_threadpool.hpp"
#include "../mqcommon/mq_proto.pb.h"
#include "mq_route.hpp"

namespace csjmq
{
    using ProtobufCodecPtr = std::shared_ptr<ProtobufCodec>;
    using openChannelRequestPtr = std::shared_ptr<openChannelRequest>;
    using closeChannelRequestPtr = std::shared_ptr<closeChannelRequest>;
    using declareExchangeRequestPtr = std::shared_ptr<declareExchangeRequest>;
    using deleteExchangeRequestPtr = std::shared_ptr<deleteExchangeRequest>;
    using declareQueueRequestPtr = std::shared_ptr<declareQueueRequest>;
    using deleteQueueRequestPtr = std::shared_ptr<deleteQueueRequest>;
    using queueBindRequestPtr = std::shared_ptr<queueBindRequest>;
    using queueUnBindRequestPtr = std::shared_ptr<queueUnBindRequest>;
    using basicPushlishRequestPtr = std::shared_ptr<basicPushlishRequest>;
    using basicAckRequestPtr = std::shared_ptr<basicAckRequest>;
    using basicConsumeRequestPtr = std::shared_ptr<basicConsumeRequest>;
    using basicCancelRequestPtr = std::shared_ptr<basicCancelRequest>;

    class Channel
    {
    public:
        using ptr = std::shared_ptr<Channel>;
        Channel(const std::string &cid, const muduo::net::TcpConnectionPtr &conn,
                const ProtobufCodecPtr &codec, const ConsumerManager::ptr &cmp,
                const VirtualHost::ptr &host, const ThreadPool::ptr &pool)
            : _cid(cid),
              _conn(conn),
              _codec(codec),
              _cmp(cmp),
              _host(host),
              _pool(pool)
        {
            DLOG("new Channel: %p", this);
        }
        ~Channel()
        {
            if (_consumer.get() != nullptr)
            {
                _cmp->remove(_consumer->tag, _consumer->qname);
            }
            DLOG("del Channel: %p", this);
        }
        // 声明和删除交换机
        void declareExchange(const declareExchangeRequestPtr &req)
        {
            bool ret = _host->declareExchange(req->exchange_name(), req->exchange_type(),
                                              req->durable(), req->auto_delete(), req->args());
            std::string rid = req->rid();
            std::string cid = req->cid();
            basicResponse(ret, rid, cid); // 将结果返回给客户端，处理结果如何
        }
        void deleteExchange(const deleteExchangeRequestPtr &req)
        {
            _host->deleteExchange(req->exchange_name());
            basicResponse(true, req->rid(), req->cid());
        }

        // 声明和删除队列
        void declareQueue(const declareQueueRequestPtr &req)
        {
            bool ret = _host->declareQueue(req->queue_name(), req->durable(),
                                           req->exclusive(), req->auto_delete(), req->args());
            if (ret == false)
            {
                return basicResponse(false, req->rid(), req->cid());
            }
            _cmp->initQueueConsumer(req->queue_name()); // 初始化队列的消费者管理句柄
            basicResponse(true, req->rid(), req->cid());
        }
        void deleteQueue(const deleteQueueRequestPtr &req)
        {
            // 删除队列时，相关的消费者也要删除
            _cmp->destoryQueueConsumer(req->queue_name());
            _host->deleteQueue(req->queue_name());
            basicResponse(true, req->rid(), req->cid());
        }

        // 绑定和解除绑定队列
        void queueBind(const queueBindRequestPtr &req)
        {
            bool ret = _host->bind(req->exchange_name(), req->queue_name(), req->binding_key());
            basicResponse(true, req->rid(), req->cid());
        }
        void queueUnbind(const queueUnBindRequestPtr &req)
        {
            bool ret = _host->unbind(req->exchange_name(), req->queue_name());
            basicResponse(true, req->rid(), req->cid());
        }

        // 发布和确认消息
        void basicPushlish(const basicPushlishRequestPtr &req)
        {
            // 1.获取交换机信息
            Exchange::ptr ep = _host->selectExchange(req->exchange_name());
            if (ep.get() == nullptr)
            {
                return basicResponse(false, req->rid(), req->cid());
            }
            // 2.进行交换路由，判断信息可以发布到交换机绑定的队列中
            MsgQueueBindingMap mqbm = _host->getExchangeBindings(req->exchange_name());
            BasicProperties *properties = nullptr; // 消息属性可能为空
            std::string routing_key;
            if (req->has_properties())
            {
                properties = req->mutable_properties();
                routing_key = properties->routing_key();
            }

            for (auto &binding : mqbm)
            {
                // 3.将信息发布到队列中（添加消息管理）
                if (Router::route(ep->type, routing_key, binding.second->binding_key))
                {
                    _host->BasicPublish(binding.first, properties, req->body());
                    // 4.向线程池中添加一个消息推送任务（向指定队列的订阅者去推送消息）
                    auto task = std::bind(&Channel::consume, this, binding.first);
                    _pool->push(task);
                }
            }
            return basicResponse(true, req->rid(), req->cid());
        }

        void basicAck(const basicAckRequestPtr &req)
        {
            _host->BasicAck(req->queue_name(), req->message_id());
            return basicResponse(true, req->rid(), req->cid());
        }

        // 订阅和取消订阅
        void basicConsume(const basicConsumeRequestPtr &req)
        {
            // 1.判断队列是否存在
            bool ret = _host->existsQueue(req->queue_name());
            if (ret == false)
            {
                basicResponse(false, req->rid(), req->cid());
                return;
            }

            // 2.创建队列的消费者
            auto cb = std::bind(&Channel::callback, this, std::placeholders::_1,
                                std::placeholders::_2, std::placeholders::_3);
            // 3.创建消费者后，当前channel角色就是个消费者标识
            _consumer = _cmp->create(req->consumer_tag(), req->queue_name(), req->auto_ack(), cb);

            return basicResponse(true, req->rid(), req->cid());
        }

        void basicCancel(const basicCancelRequestPtr &req)
        {
            _cmp->remove(req->consumer_tag(), req->queue_name());
            return basicResponse(true, req->rid(), req->cid());
        }

    private:
        void callback(const std::string &tag, const BasicProperties *bp, const std::string body)
        {
            basicConsumeResponse resp;
            resp.set_cid(_cid);
            resp.set_body(body);
            resp.set_consumer_tag(tag);
            if (bp)
            {
                resp.mutable_properties()->set_id(bp->id());
                resp.mutable_properties()->set_delivery_mode(bp->delivery_mode());
                resp.mutable_properties()->set_routing_key(bp->routing_key());
            }
            _codec->send(_conn, resp);
        }

        void basicResponse(bool ok, const std::string &rid, const std::string &cid)
        {
            basicCommonResponse resp;
            resp.set_rid(rid);
            resp.set_cid(cid);
            resp.set_ok(ok);
            _codec->send(_conn, resp);
        }
        void consume(const std::string &qname)
        {
            // 1.从队列中取出一条消息
            MessagePtr mp = _host->BasicConsume(qname);
            if (mp.get() == nullptr)
            {
                DLOG("执行消费任务失败, %s队列不存在", qname.c_str());
                return;
            }

            // 2.从队列订阅者中取出一个订阅者
            Consumer::ptr cp = _cmp->choose(qname);
            if (cp.get() == nullptr)
            {
                DLOG("执行消费任务失败, %s队列没有订阅者", qname.c_str());
            }

            // 3.调用订阅者对应的消息回调函数，实现消息推送
            cp->callback(cp->tag, mp->mutable_payload()->mutable_properties(), mp->payload().body());

            // 4.判断如果订阅者是自动确认---不需要等待确认，直接删除消息，否则需要外部收到消息确认后再删除
            if (cp->auto_ack)
                _host->BasicAck(qname, mp->payload().properties().id());
        }

    private:
        std::string _cid;                   // 信道id
        Consumer::ptr _consumer;            // 关联的消费者
        muduo::net::TcpConnectionPtr _conn; // 关联的连接
        ProtobufCodecPtr _codec;            // protobuf的协议处理
        ConsumerManager::ptr _cmp;          // 消费者管理句柄
        VirtualHost::ptr _host;             // 虚拟机句柄（交换机，队列，绑定信息，消息数据）
        ThreadPool::ptr _pool;              // 线程池句柄
    };

    class ChannelManager
    {
    public:
        using ptr = std::shared_ptr<ChannelManager>;
        ChannelManager() {}
        bool openChannel(const std::string &cid, const muduo::net::TcpConnectionPtr &conn,
                         const ProtobufCodecPtr &codec, const ConsumerManager::ptr &cmp,
                         const VirtualHost::ptr &host, const ThreadPool::ptr &pool)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _channels.find(cid);
            if (it != _channels.end())
            {
                DLOG("%s 信道已经存在", cid.c_str());
                return false;
            }
            _channels[cid] = std::make_shared<Channel>(cid, conn, codec, cmp, host, pool);
            return true;
        }

        bool closeChannel(const std::string &cid)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _channels.find(cid);
            if (it == _channels.end())
            {
                DLOG("%s 信道不存在", cid.c_str());
                return false;
            }
            _channels.erase(cid);
            return true;
        }

        Channel::ptr getChannel(const std::string &id)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _channels.find(id);
            if (it == _channels.end())
            {
                return Channel::ptr();
            }
            return it->second;
        }

    private:
        std::mutex _mutex;
        std::unordered_map<std::string, Channel::ptr> _channels;
    };
}

#endif